package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo3CarFlow {
  def main(args: Array[String]): Unit = {
    /**
     * 1、创建flink sql环境
     */
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() //流处理模式
      //.inBatchMode()//批处理模式
      .build()

    val tableENV: TableEnvironment = TableEnvironment.create(settings)

    /**
     * 1、创建kafka source表
     */
    tableENV.executeSql(
      """
        |CREATE TABLE kafka_cars (
        |    car STRING, --  车牌号
        |    city_code STRING,-- 城市编号
        |    county_code STRING, -- 区县编号
        |    card BIGINT, -- 卡口编号
        |    camera_id STRING, -- 摄像头编号
        |    orientation STRING,-- 方向
        |    road_id BIGINT, -- 道路编号
        |    `time` BIGINT, -- 时间
        |    speed DOUBLE, -- 车速
        |    proc_time as PROCTIME() -- 增加处理时间字段
        |) WITH (
        |  'connector' = 'kafka', -- 数据源类型
        |  'topic' = 'cars', -- topic
        |  'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- kafka broker列表
        |  'properties.group.id' = 'testGroup', -- 消费者组
        |  'scan.startup.mode' = 'earliest-offset', -- 读取数据位置
        |  'format' = 'json' -- 读取数据的格式，json格式会按照字段名映射
        |)
        |""".stripMargin)

    /**
     * 2、创建sink 表
     */
    tableENV.executeSql(
      """
        |CREATE TABLE card_flow_avg_speed (
        |    card BIGINT,
        |    win_end TIMESTAMP(3),
        |    flow BIGINT,
        |    avg_speed DOUBLE,
        |    PRIMARY KEY (card,win_end) NOT ENFORCED -- 增加主键，flink会通过主键更新数据
        |) WITH (
        |    'connector' = 'jdbc', -- 数据源类型，jdbc
        |    'url' = 'jdbc:mysql://master:3306/bigdata', -- jdbc url
        |    'table-name' = 'card_flow_avg_speed', -- mysql中的表名，需要提前创建号
        |    'username' = 'root',
        |    'password' = '123456'
        |)
        |""".stripMargin)

    /**
     * 在动态表上计算连续查询将查询结果插入到sink表中
     */
    tableENV.executeSql(
      """
        |insert into card_flow_avg_speed
        |select
        |    card,
        |    HOP_END(proc_time, INTERVAL '5' SECOND, INTERVAL '15' SECOND) as win_end ,-- 窗口结束时间
        |    count(1) as flow, -- 车流量
        |    avg(speed) as avg_speed -- 平均车速
        |from
        |	kafka_cars
        |group by
        |    card,
        |    HOP(proc_time, INTERVAL '5' SECOND, INTERVAL '15' SECOND)
        |
        |""".stripMargin)

  }

}
